package com.bw.yk01;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

//sku_info>>>>> {"op":"c","data":{"is_sale":0,"sku_desc":"CAREMiLLE珂曼奶油小方口红 雾面滋润保湿持久丝缎唇膏 M02干玫瑰","tm_id":9,"create_time":"2020-11-18 23:23:11","price":69,"sku_default_img":"http://47.93.148.192:8080/group1/M00/00/02/rBHu8l-0y1eASPU7AAEiYRrkMww103.jpg","weight":1.00,"sku_name":"CAREMiLLE珂曼奶油小方口红 雾面滋润保湿持久丝缎唇膏 M02干玫瑰","id":30,"spu_id":10,"category3_id":477},"db":"gmall","tb":"sku_info"}
//{op  data  c db}
public class DimRedisSink extends RichSinkFunction<String> {

    Jedis jedis = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = new Jedis("hadoop-single");
    }

    @Override
    public void close() throws Exception {
        jedis.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        //解析出表名tb 作为key
        //解析出data 作为value
        String tb = JSON.parseObject(value).getString("tb");
        JSONObject data = JSON.parseObject(value).getJSONObject("data");
        String id = data.getString("id");
        jedis.set(tb+":"+id,data.toString());

    }
}
